package com.tdy.cdc.app.cdr;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamStatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: NanHuang
 * @Date: 2023/04/10/17:40
 * @Description: DWD层：dwd_mhs_test_report 表手动分流
 */
public class Cdr_Hdsd00_05_01 extends BaseSqlApp {
    public static void main(String[] args) {
        new Cdr_Hdsd00_05_01().init(
                6006,
                "Cdr_Hdsd00_05_01",
                2
        );
    }
    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        super.readPatient(tableEnv);
        super.readVisit(tableEnv);

        tableEnv.executeSql("create table dwd_mhs_test_report (" +
                "XDS_TYPE                       STRING, " +
                "XDS_NAME                       STRING, " +
                "XDS_VERSION                    STRING, " +
                "PATIENT_DOMAIN                 STRING, " +
                "VISIT_DOMAIN                   STRING, " +
                "CUSTODIAN_CODE                 STRING, " +
                "CUSTODIAN_NAME                 STRING, " +
                "HOSPITAL_CODE                  STRING, " +
                "HOSPITAL_NAME                  STRING, " +
                "PATIENT_ID                     STRING, " +
                "INPATIENT_VISIT_FLOW_ID        STRING, " +
                "OUTPATIENT_VISIT_FLOW_ID       STRING, " +
                "VISIT_TIMES                    STRING, " +
                "RECORD_ID                      STRING, " +
                "DW_DATE                        STRING, " +
                "OUTPATIENT_NO                  STRING, " +
                "DOMAIN_ID                      STRING, " +
                "VISIT_DATE                     STRING, " +
                "INPATIENT_NO                   STRING, " +
                "PATIENT_VISIT_CATEGORY_CODE    STRING, " +
                "PATIENT_VISIT_CATEGORY         STRING, " +
                "TEST_REPORT_NO                 STRING, " +
                "APPLY_SHEET_NO                 STRING, " +
                "SPECIMEN_NUMBER                STRING, " +
                "TELECOM                        STRING, " +
                "ID_CARD_NUMBER                 STRING, " +
                "PATIENT_NAME                   STRING, " +
                "SEX_CODE                       STRING, " +
                "SEX_NAME                       STRING, " +
                "AGE                            STRING, " +
                "AGE_UNIT                       STRING, " +
                "TEST_DATE                      STRING, " +
                "TEST_REPORT_DOCTOR_ID          STRING, " +
                "TEST_REPORT_DOCTOR_NAME        STRING, " +
                "AUDIT_SIGNDATE                 STRING, " +
                "TEST_AUDIT_DOCTOR_ID           STRING, " +
                "TEST_AUDIT_DOCTOR_NAME         STRING, " +
                "CHECK_SIGNDATE                 STRING, " +
                "TEST_TECH_STAFF_ID             STRING, " +
                "TEST_TECH_STAFF_NAME           STRING, " +
                "TECHNICIAN_SIGNDATE            STRING, " +
                "TEST_DOCTOR_ID                 STRING, " +
                "TEST_DOCTOR_NAME               STRING, " +
                "APPLY_DATE                     STRING, " +
                "APPLY_DEPT_ID                  STRING, " +
                "APPLY_DEPT_NAME                STRING, " +
                "APPLY_MED_ORG_ID               STRING, " +
                "TEST_REPORT_DEPT_ID            STRING, " +
                "BED_NO                         STRING, " +
                "BED_NAME                       STRING, " +
                "ROOM_NO                        STRING, " +
                "ROOM_NAME                      STRING, " +
                "DEPT_ID                        STRING, " +
                "DEPT_NAME                      STRING, " +
                "WARDS_ID                       STRING, " +
                "WARDS_NAME                     STRING, " +
                "INSPECTION_METHOD              STRING, " +
                "INSPECT_TYPE                   STRING, " +
                "TEST_REPORT_RESULT             STRING, " +
                "REPORT_DEPT                    STRING, " +
                "AGENCY_NAME                    STRING, " +
                "TEST_REPORT_COOMMENTS          STRING, " +
                "INSPECT_TYPE_CODE              STRING, " +
                "INSPECTION_ITEM                STRING, " +
                "EXAM_DATE                      STRING, " +
                "ITEM_CODE                      STRING, " +
                "ITEM_NAME                      STRING, " +
                "SPECIMEN_SAMPLE_TIME           STRING, " +
                "SPECIMEN_RECV_TIME             STRING, " +
                "SPECIMEN_CLASS                 STRING, " +
                "SPECIMEN_STATUS                STRING, " +
                "SPECIMEN_FIXATIVE_NAME         STRING, " +
                "RESULT_NAME                    STRING, " +
                "RESULT_CODE                    STRING, " +
                "EXAM_RESULT_VALUE              STRING, " +
                "EXAM_RESULT_UNIT               STRING, " +
                "DIAG_DATE                      STRING, " +
                "DIAG_CODE                      STRING, " +
                "DIAG_NAME                      STRING, " +
                "DIAG_TYPE                      STRING, " +
                "DIAG_INSTITUTION               STRING, " +
                "`pt` as proctime()" +
                ")" +
                FlinkSourceUtil.getKafkaWith(Constant.TOPIC_DWD_MHS_TEST_REPORT,"test"));
        // cdr_hdsd00_05_01
        tableEnv.executeSql("create table cdr_hdsd00_05_01 (" +
                "   xds_type              string," +
                "   xds_name              string," +
                "   xds_version           string," +
                "   patient_domain        string," +
                "   visit_domain          string," +
                "   custodian_code        string," +
                "   custodian_name        string," +
                "   hospital_code         string," +
                "   hospital_name         string," +
                "   patient_id            string," +
                "   visit_id              string," +
                "   visit_times           string," +
                "   unique_id             string," +
                "   effective_time        string," +
                "   out_patient_id        string," +
                "   domain_id             string," +
                "   visit_date            string," +
                "   inpatient             string," +
                "   patient_typecode      string," +
                "   patient_type          string," +
                "   report_id             string," +
                "   appy_id               string," +
                "   specimen_number       string," +
                "   telecom               string," +
                "   identify_no           string," +
                "   patient_name          string," +
                "   sex_code              string," +
                "   sex                   string," +
                "   age                   string," +
                "   age_unit              string," +
                "   write_time            string," +
                "   author_code           string," +
                "   author_name           string," +
                "   audit_signdate        string," +
                "   audit_code            string," +
                "   audit_name            string," +
                "   check_signdate        string," +
                "   technician_code       string," +
                "   technician_name       string," +
                "   technician_signdate   string," +
                "   examiner_code         string," +
                "   examiner_name         string," +
                "   apply_date            string," +
                "   apply_dept_code       string," +
                "   apply_dept            string," +
                "   apply_agencycode      string," +
                "   apply_organ           string," +
                "   bed_no                string," +
                "   bed_name              string," +
                "   ward_id               string," +
                "   ward_name             string," +
                "   dept_code             string," +
                "   dept_name             string," +
                "   wards_id              string," +
                "   wards_name            string," +
                "   inspection_method     string," +
                "   inspect_type          string," +
                "   report_result         string," +
                "   report_dept           string," +
                "   agency_name           string," +
                "   report_remark         string," +
                "   inspect_type_code     string," +
                "   inspection_item       string," +
                "   status                string," +
                "   empi                  string," +
                "   encounter_id          string " +
                ")" +
                FlinkSinkUtil.getKafkaWith("cdr_hdsd00_05_01"));
        // cdr_tdy_list_diag
        tableEnv.executeSql("create table cdr_tdy_list_diag(" +
                "unique_id              string," +
                "xds_type               string," +
                "xds_version            string," +
                "domain_id              string," +
                "diag_code              string," +
                "diag_name              string," +
                "diag_type              string," +
                "diag_institution       string " +
                ")" +
                FlinkSinkUtil.getKafkaWith("cdr_tdy_list_diag"));

        // cdr_tdy_list_test
        tableEnv.executeSql("create table cdr_tdy_list_test(" +
                "unique_id                string," +
                "xds_type                 string," +
                "xds_version              string," +
                "domain_id                string," +
                "inspect_date             string," +
                "item_name                string," +
                "item_code                string," +
                "sample_date              string," +
                "receipt_date             string," +
                "result_code              string," +
                "result_name              string," +
                "quanty_result            string," +
                "quanty_unit              string," +
                "reference_value          string," +
                "result_sign              string," +
                "quality_name             string " +
                ")" +
                FlinkSinkUtil.getKafkaWith("cdr_tdy_list_test"));

        StreamStatementSet statementSet = tableEnv.createStatementSet();
        statementSet.addInsertSql("insert into cdr_hdsd00_05_01 " +
                        "select " +
                        "   'C0007'                           ," +
                        "   '检查报告'                         ," +
                        "   'v1.0.0.0'                        ," +
                        "   t.PATIENT_DOMAIN                  ," +
                        "   t.VISIT_DOMAIN                    ," +
                        "   t.CUSTODIAN_CODE                  ," +
                        "   t.CUSTODIAN_NAME                  ," +
                        "   t.HOSPITAL_CODE                   ," +
                        "   t.HOSPITAL_NAME                   ," +
                        "   t.PATIENT_ID                      ," +
                        "   case when t.PATIENT_VISIT_CATEGORY_CODE='3' then t.INPATIENT_VISIT_FLOW_ID else t.OUTPATIENT_VISIT_FLOW_ID end                         ," +
                        "   t.VISIT_TIMES                     ," +
                        "   t.TEST_REPORT_NO                  ," +
                        "   t.DW_DATE                         ," +
                        "   t.OUTPATIENT_NO                   ," +
                        "   t.DOMAIN_ID                       ," +
                        "   t.VISIT_DATE                      ," +
                        "   t.INPATIENT_NO                    ," +
                        "   t.PATIENT_VISIT_CATEGORY_CODE     ," +
                        "   t.PATIENT_VISIT_CATEGORY          ," +
                        "   t.TEST_REPORT_NO                  ," +
                        "   t.APPLY_SHEET_NO                  ," +
                        "   t.SPECIMEN_NUMBER                 ," +
                        "   t.TELECOM                         ," +
                        "   t.ID_CARD_NUMBER                  ," +
                        "   t.PATIENT_NAME                    ," +
                        "   t.SEX_CODE                        ," +
                        "   t.SEX_NAME                        ," +
                        "   t.AGE                             ," +
                        "   t.AGE_UNIT                        ," +
                        "   t.TEST_DATE                       ," +
                        "   t.TEST_REPORT_DOCTOR_ID           ," +
                        "   t.TEST_REPORT_DOCTOR_NAME         ," +
                        "   t.AUDIT_SIGNDATE                  ," +
                        "   t.TEST_AUDIT_DOCTOR_ID            ," +
                        "   t.TEST_AUDIT_DOCTOR_NAME          ," +
                        "   t.CHECK_SIGNDATE                  ," +
                        "   t.TEST_TECH_STAFF_ID              ," +
                        "   t.TEST_TECH_STAFF_NAME            ," +
                        "   t.TECHNICIAN_SIGNDATE             ," +
                        "   t.TEST_DOCTOR_ID                  ," +
                        "   t.TEST_DOCTOR_NAME                ," +
                        "   t.APPLY_DATE                      ," +
                        "   t.APPLY_DEPT_ID                   ," +
                        "   t.APPLY_DEPT_NAME                 ," +
                        "   t.APPLY_MED_ORG_ID                ," +
                        "   t.TEST_REPORT_DEPT_ID             ," +
                        "   t.BED_NO                          ," +
                        "   t.BED_NAME                        ," +
                        "   t.ROOM_NO                         ," +
                        "   t.ROOM_NAME                       ," +
                        "   t.DEPT_ID                         ," +
                        "   t.DEPT_NAME                       ," +
                        "   t.WARDS_ID                        ," +
                        "   t.WARDS_NAME                      ," +
                        "   t.INSPECTION_METHOD               ," +
                        "   t.INSPECT_TYPE                    ," +
                        "   t.TEST_REPORT_RESULT              ," +
                        "   t.REPORT_DEPT                     ," +
                        "   t.AGENCY_NAME                     ," +
                        "   t.TEST_REPORT_COOMMENTS           ," +
                        "   t.INSPECT_TYPE_CODE               ," +
                        "   t.INSPECTION_ITEM                 ," +
                        "   if(p.PATIENT_ID is not null and (v1.VISIT_ID is not null or v2.VISIT_ID is not null),'AA','AE')            ," +
                        "   p.GLOBAL_ID                                                                                                ," +
                        "   case " +
                        "      when v1.VISIT_ID is not null then v1.VISIT_DOMAIN_ID || '_' || v1.VISIT_ID || '_' || v1.PATIENT_CATEGORY " +
                        "      when v2.VISIT_ID is not null then v2.VISIT_DOMAIN_ID || '_' || v2.VISIT_ID || '_' || v2.PATIENT_CATEGORY " +
                        "   end  " +
                        "from dwd_mhs_test_report t                                                       " +
                        "left join patient for system_time as of t.pt as p on t.PATIENT_ID = p.PATIENT_ID " +
                        "left join visit for system_time as of t.pt as v1 on t.INPATIENT_VISIT_FLOW_ID = v1.VISIT_ID        " +
                        "left join visit for system_time as of t.pt as v2 on t.OUTPATIENT_VISIT_FLOW_ID = v2.VISIT_ID       ")
                .addInsertSql("insert into cdr_tdy_list_diag " +
                        "select " +
                        "   TEST_REPORT_NO        ," +
                        "   'C0007'               ," +
                        "   'v1.0.0.0'            ," +
                        "   DOMAIN_ID             ," +
                        "   DIAG_CODE             ," +
                        "   DIAG_NAME             ," +
                        "   DIAG_TYPE             ," +
                        "   DIAG_INSTITUTION       " +
                        "from dwd_mhs_test_report  " +
                        "where DIAG_CODE is not null and DIAG_CODE <> '' ")
                .addInsertSql("insert into cdr_tdy_list_test " +
                        "select " +
                        "   TEST_REPORT_NO        ," +
                        "   'C0007'               ," +
                        "   'v1.0.0.0'            ," +
                        "   DOMAIN_ID             ," +
                        "   INSPECT_DATE          ," +
                        "   ITEM_NAME             ," +
                        "   ITEM_CODE             ," +
                        "   SAMPLE_DATE           ," +
                        "   RECEIPT_DATE          ," +
                        "   TEST_RESULT_CODE      ," +
                        "   TEST_RESULT_NAME      ," +
                        "   TEST_RESULT_VALUE     ," +
                        "   TEST_RESULT_UNIT      ," +
                        "   TEST_REFERENCE_RANGE  ," +
                        "   RESULT_SIGN           ," +
                        "   TEST_RESULT_QUALITATIVE_VALUE " +
                        "from dwd_mhs_test_report  " +
                        "where ITEM_CODE is not null and ITEM_CODE <> '' ");
        statementSet.execute();
    }
}
